package org.budo.redis.jdbc.driver.executor.socket;

import java.io.FilterInputStream;
import java.io.InputStream;

import org.budo.support.java.io.util.InputStreamUtil;
import org.budo.support.lang.util.ByteArray;

/**
 * @author lmw
 */
public class BudoRedisSocketInputStream extends FilterInputStream {
    private static final int BUF_SIZE_DEFAULT = 8192;

    private final byte[] buf;

    private int bufIndex, bufLen;

    public BudoRedisSocketInputStream(InputStream inputStream) {
        super(inputStream);

        buf = new byte[BUF_SIZE_DEFAULT];
    }

    private void fillBuf() {
        if (bufIndex >= bufLen) {
            bufLen = InputStreamUtil.read(this.in, buf);
            bufIndex = 0;

            if (bufLen == -1) {
                throw new IllegalStateException("#180 Unexpected end of stream.");
            }
        }
    }

    @Override
    public int read(byte[] to, int off, int len) {
        this.fillBuf();

        int length = Math.min(bufLen - bufIndex, len);
        System.arraycopy(buf, bufIndex, to, off, length);
        bufIndex += length;
        return length;
    }

    public byte readByte() {
        this.fillBuf();

        return buf[bufIndex++];
    }

    public long readCRLF() {
        byte[] buf = this.buf;

        this.fillBuf();

        boolean isNeg = buf[bufIndex] == '-';
        if (isNeg) {
            ++bufIndex;
        }

        long value = 0;
        while (true) {
            this.fillBuf();

            int b = buf[bufIndex++];
            if (b == '\r') {
                this.fillBuf();

                if (buf[bufIndex++] != '\n') {
                    throw new IllegalStateException("#152 Unexpected character!");
                }

                break;
            }

            value = value * 10 + b - '0';
        }

        return (isNeg ? -value : value);
    }

    public byte[] readString() {
        ByteArray byteArray = new ByteArray();

        while (true) {
            this.fillBuf();

            byte b1 = buf[bufIndex++];
            if (b1 == '\r') {
                this.fillBuf(); // Must be one more byte

                byte b2 = buf[bufIndex++];
                if (b2 == '\n') {
                    break;
                }

                byteArray.write(b1);
                byteArray.write(b2);
            } else {
                byteArray.write(b1);
            }
        }

        return byteArray.getBytes();
    }

    protected byte[] readBulkString() {
        int arrayLen = (int) this.readCRLF();

        if (arrayLen == -1) {
            return null;
        }

        byte[] buf = new byte[arrayLen];
        int offset = 0;
        while (offset < arrayLen) {
            final int size = this.read(buf, offset, (arrayLen - offset));
            if (size == -1) {
                throw new IllegalStateException("It seems like server has closed the connection.");
            }
            offset += size;
        }

        // read 2 more bytes for the command delimiter
        this.readByte();
        this.readByte();

        return buf;
    }
}